package org.apache.hadoop.mapred;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.LinkedList;
import java.util.Map;
import java.util.Set;

public class Graph {

public static class Edge {

	int num_flows;
	int dummy;//dummy for computing metric. compare dummy and current e.g. num_flows
	double bandwidth;
	//for traversing edges
	boolean edge_visited;

	public Edge() {
		this.num_flows = 0;
		this.dummy = 0;
		this.bandwidth = 0.0;
		this.edge_visited = false;
	}

	public Edge(int numFlows, int dum, double bw, boolean edge_flag) {
		this.num_flows = numFlows;
		this.dummy = dum;
		this.bandwidth = bw;
		this.edge_visited = edge_flag;
	}
}

public static class Vertex {

	Map<String, Edge> neighbour = new HashMap<String, Edge>();

	String nodename;
	int type; //switch or node

	//for compute path
	int dist;
	Vertex path;

	//for traversing edges
	boolean vertex_visited;
}


	Map<String, Vertex> topology = new HashMap<String, Vertex>();//topology map<nodename, vertex*>
	Map<String, Map<String, ArrayList<String> > > path = new HashMap<String, Map<String, ArrayList<String> > >();//path table (map<src, map<dst, node_on_the_path>>)

	int num_worknodes; //only work nodes
	String jobtracker;


	public void graph_add_edge(String node1, int type1, String node2, int type2, double bw) {
	
		//Map<String, Vertex> topology
		Vertex n1;
		Vertex n2;
		
		n1 = topology.get(node1);
		n2 = topology.get(node2);
		
		if(n1 == null)
		{
			n1 = new Vertex();
			n1.nodename = node1;
			n1.type = type1;
			topology.put(node1, n1);			
		}

		if(n2 == null)
		{
			n2 = new Vertex();
			n2.nodename = node2;
			n2.type = type2;
			topology.put(node2, n2);			
		}
			
		Edge edge = n1.neighbour.get(node2);
		if(edge != null)
			throw new IllegalArgumentException();
			
		edge = new Edge();
		
		edge.bandwidth = bw;
		
		n1.neighbour.put(node2, edge);
		
		
		Edge edge2 = n2.neighbour.get(node1);
		if(edge2 != null)
			throw new IllegalArgumentException();
			
		edge2 = new Edge();
		
		edge2.bandwidth = bw;
		
		n2.neighbour.put(node1, edge2);

	}
	
	
	public void compute_path() {
	
		//Map<String, Vertex> topology
		for(String nodename : topology.keySet())
		{
			Vertex v = topology.get(nodename);
			if(v.type == 1)
			{
				if(nodename.equals(jobtracker))
					System.out.print("Graph::compute_path() Job tracker is: "+jobtracker+"\n");
				else
					num_worknodes++;//count task nodes in the cluster
					
				
				shortest_path(nodename);
				
				//	Map<String, Map<String, ArrayList<String> > > path //path table (map<src, map<dst, node_on_the_path>>)
				Map<String, ArrayList<String> > src_path = new HashMap<String, ArrayList<String> >();
				
				for(String nodename2 : topology.keySet())
				{
					Vertex v2 = topology.get(nodename2);
					if(v2.type == 1 && !nodename2.equals(nodename))//dst is node and not the same as src, not switch
					{
						ArrayList<String> path = new ArrayList<String>();
						
						print_path(nodename2, path);
						
						src_path.put(nodename2, path);
					
					}
				}
				
				path.put(nodename, src_path);
				
			}//if(v.type == 1)
		}
		
		//test path from n_rg0_0_ng0_0 to n_rg0_1_ng0_1
		/*		
		Map<String, ArrayList<String> > test1 = path.get("4");
		if(test1 == null)
			throw new NullPointerException("test1");
		ArrayList<String> test2 = test1.get("1");
		if(test2 == null)
			throw new NullPointerException("test2");
		System.out.print("test2 size:"+test2.size()+"\n");		
		for(String test3 : test2)
			System.out.print(test3+"  ");
		System.out.print("\n");
		*/
	}
	
	//compute the shortest path
	private void shortest_path(String src) {
		System.out.print("compute shortest_path from "+src+"\n");
		
		LinkedList<String> q = new LinkedList<String> ();
	
		//Map<String, Vertex> topology
		for(Vertex vertex : topology.values()) {
			vertex.path = null;
			vertex.dist = Integer.MAX_VALUE;
		}

		Vertex sv = topology.get(src);
		if(sv == null)
			throw new NullPointerException();

		sv.dist = 0;
		
		q.addLast(src);
		
		while(!q.isEmpty())
		{
			String v = q.removeFirst();
		
			Vertex pv = topology.get(v);
			if(pv == null)
				throw new NullPointerException();
			
			//Map<String, Edge> neighbour
			for(String eit : pv.neighbour.keySet()) {
				Vertex adjacent = topology.get(eit);
				if(adjacent == null)
					throw new NullPointerException();
	
				if(adjacent.dist == Integer.MAX_VALUE)
				{
					adjacent.dist = pv.dist + 1;
					adjacent.path = pv;
			
					q.addLast(adjacent.nodename);
				}
			}
		}//while
	}
	
	
	private void print_path(String dst, ArrayList<String> path) {
		
		Vertex pv = topology.get(dst);
		if(pv == null)
			throw new NullPointerException();
			
		if(pv.path != null) {
			print_path(pv.path.nodename, path);
			//System.out.print();
		}
		
		path.add(pv.nodename);

	}

	//Breadth-First-Search variant and perform particular action on each edge	
	//Map<String, Vertex> topology //topology map<nodename, vertex*>
	//Map<String, Map<String, ArrayList<String> > > path //path table (map<src, map<dst, node_on_the_path>>)
	public void traverse_edge(String action, ArrayList<Double> edge_values) {
	
		//i ==0: reset edge_visited to false first. then do the action
		for(int i=0; i<2; i++)
		{
			LinkedList<String> q = new LinkedList<String> ();
			
			boolean first = true;
			for(String node:topology.keySet())
			{
				Vertex v = topology.get(node);
				if(first == true)
				{
					q.addLast(node);
					v.vertex_visited = true;
					
					first = false;
				}
				
				v.vertex_visited = false;
			
			}
		
			while(!q.isEmpty())
			{
				String v = q.removeFirst();
				Vertex pv = topology.get(v);

				//Map<String, Edge> neighbour
				for(String eit : pv.neighbour.keySet()) {
					Vertex adjacent = topology.get(eit);
					
					Edge forward = pv.neighbour.get(adjacent.nodename);//forward edge
					Edge reverse = adjacent.neighbour.get(v);//reverse edge
	
					if( i == 0) {
						forward.edge_visited = reverse.edge_visited = false;
					} else if(forward.edge_visited == false) {
					
						if(reverse.edge_visited != false)
							throw new IllegalArgumentException("reverse.edge_visited != false");
						
						if (action.equals("get_flow_dummy")) {
							
							if(forward.dummy != reverse.dummy)
								throw new IllegalArgumentException();
								
							//normalize the flow here
							double normal_dummy = (double)forward.dummy/forward.bandwidth;
						
							edge_values.add(normal_dummy);
						
							/*
							cout<<"edge_values.push_back "<<" src: "<<eit->first<<" dst: "<<eit2->first<<endl;
							cout<<"edge_values.push_back "<<" src: "<<eit->second->dummy<<" dst: "<<eit->second->dummy<<endl;
							cout<<"edge_values.push_back "<<" src: "<<eit->second->num_flows<<" dst: "<<eit->second->num_flows<<endl;
							*/
						} else if (action.equals("set_flow_dummy")) {
						
							if(forward.dummy != reverse.dummy)
								throw new IllegalArgumentException();						
						
							if(forward.num_flows != reverse.num_flows)
								throw new IllegalArgumentException();
								
							forward.dummy = reverse.dummy = forward.num_flows;

						}
						
						forward.edge_visited = reverse.edge_visited = true;

					}

					if(adjacent.vertex_visited == false)
					{
						adjacent.vertex_visited = true;
						q.addLast(adjacent.nodename);
					}

				}
			}//while
		}//for i	
	}
	
	//perform action on particular variable
	//action = start/finish
	public void flow_change(String src, String dst, String action) {

		if(src.equals(dst))
			throw new IllegalArgumentException();

		//Map<String, Map<String, ArrayList<String> > > path //path table (map<src, map<dst, node_on_the_path>>)
		Map<String, ArrayList<String> > dst_map = path.get(src);
		ArrayList<String> flow_path = dst_map.get(dst);
		if(flow_path.size() < 2)
			throw new IllegalArgumentException();
			
		int it1 = 0;
		int it2 = 1;
		
		do {
			Vertex v1 = topology.get(flow_path.get(it1));
			Vertex v2 = topology.get(flow_path.get(it2));
			
			//undirected graph, symmetric
			Edge edge1 = v1.neighbour.get(flow_path.get(it2));
			Edge edge2 = v2.neighbour.get(flow_path.get(it1));
		
			if(action.equals("start"))
			{
				//cout<<"start prev flows: "<< edge1->num_flows << " == "<< edge2->num_flows<<endl;
				edge1.num_flows += 1;
				edge2.num_flows += 1;
			}
			else if(action.equals("finish"))
			{
				//cout<<"finish prev flows: "<< edge1->num_flows << " == "<< edge2->num_flows<<endl;
				edge1.num_flows -= 1;
				edge2.num_flows -= 1;
			}
			else if(action.equals("add_flow_dummy"))
			{
				//cout<<"add prev dummy: "<< edge1->dummy << " == "<< edge2->dummy<<endl;
				//assert(edge1->num_flows == edge2->num_flows && edge1->dummy == edge1->num_flows && edge2->dummy == edge2->num_flows);//not hold for reduce multi-flows
				
				if(edge1.num_flows != edge2.num_flows)
					throw new IllegalArgumentException();
					
				if(edge1.dummy != edge2.dummy)
					throw new IllegalArgumentException();					
			
				edge1.dummy += 1;
				edge2.dummy += 1;

				//cout<<"add_flow_dummy "<<" src: "<<*it1<<" dst: "<<*it2<<endl;
			}
			else
				throw new IllegalArgumentException();
				
			it1++;
			it2++;		
		
		} while(it2 < flow_path.size());

	}

	public double min_std_flows_metric(String src, String dst)
	{
	//	cout<<"min_flows_metric "<<src<<":"<<dst<<endl;
	
		if(src.equals(dst))
			throw new IllegalArgumentException();
	
		ArrayList<Double> edge_values = new ArrayList<Double>();
	
		//set dummy to be num of flows
		traverse_edge("set_flow_dummy", edge_values);
	
		flow_change(src, dst, "add_flow_dummy");
	
		//get NORMALIZED dummy after adding flow
		traverse_edge("get_flow_dummy", edge_values);
	
		double metric = standard_deviation_compute(edge_values);

		return metric;
	}

	public double min_flows_per_link_metric(String src, String dst)
	{
	//	cout<<"min_flows_metric "<<src<<":"<<dst<<endl;
	
		if(src.equals(dst))
			throw new IllegalArgumentException();
				
		ArrayList<Double> edge_values = new ArrayList<Double>();
	
		//set dummy to be num of flows
		traverse_edge("set_flow_dummy", edge_values);
	
		flow_change(src, dst, "add_flow_dummy");
	
		//get NORMALIZED dummy after adding flow
		traverse_edge("get_flow_dummy", edge_values);
	
		//get max # of flows/link in the cluster
		double metric = Collections.max(edge_values);
		
		return metric;
	}
	
	
	static double sum_compute(ArrayList<Double> values)
	{
		double sum =0.0;
		for(Double it:values)	
			sum += it;
		return sum;
	}

	static double standard_deviation_compute(ArrayList<Double> values)
	{
		double total = sum_compute(values);
		double average = total/( (double)values.size() );
	
		ArrayList<Double> deviation = new ArrayList<Double>();

		for(Double it:values)
		{
			double x = Math.pow(it-average, 2.0 );
			deviation.add(x);
		}
	
		double deviation_total = sum_compute(deviation);
	
		double sd = Math.sqrt(deviation_total/( (double)values.size() ));

		//cout<<"sd is "<<values.size()<<";"<<total<<":"<<sd<<endl;

		return sd;	
	}	





/*
	public static void main(String[] args) {
	
		System.out.print("Hello world!\n");
		
		Graph g = new Graph();
		
		//specify jt of graph g
		g.jobtracker = "2";
		 
		//System.out.print(g.jobtracker+"\n"); 
	
		//public void graph_add_edge(String node1, int type1, String node2, int type2, double bw) 
		//type 1: nodes (jt or tt) 0: switches
		g.graph_add_edge("1", 1, "5", 1, 1.0);
		g.graph_add_edge("3", 1, "5", 1, 1.0);
		g.graph_add_edge("5", 1, "6", 1, 1.0);
		g.graph_add_edge("6", 1, "7", 1, 1.0);
		g.graph_add_edge("7", 1, "2", 1, 1.0);
		g.graph_add_edge("4", 1, "7", 1, 1.0);
			 	 

		g.compute_path();	 
	
		System.out.print("finish!"+g.num_worknodes+"\n");
		
		
		//flow_change(String src, String dst, String action) {
		//add to Edge.num_flows (which is actual flow)
		g.flow_change("2", "3", "start");
		g.flow_change("1", "4", "start");
		
		//traverse_edge(String action, ArrayList<Double> edge_values) {
		ArrayList<Double> edge_values = new ArrayList<Double>();
	
		//set dummy to Edge.num_flows
		g.traverse_edge("set_flow_dummy", edge_values);
	
		
		//g.flow_change("3", "4", "add_flow_dummy");//add to dummy! no effect on actual num_flows!
		double dou = g.min_flows_per_link_metric("3","4");//return max flows on edge in graph!
		System.out.println("dou:"+dou);
	
		//get NORMALIZED dummy after adding flow
		g.traverse_edge("get_flow_dummy", edge_values);//get the value of Edge.dummy on link
		
		for(Double f:edge_values) {
		
			System.out.println("edge: "+f);
		
		}

		System.out.println("------------------");
//remove
		edge_values.clear();
		
		////remove Edge.num_flows (which is actual flow)
		g.flow_change("2", "3", "finish");
		

		//set dummy to be num of flows
		g.traverse_edge("set_flow_dummy", edge_values);
	
		//flow_change(src, dst, "add_flow_dummy");
	
		//get NORMALIZED dummy after adding flow
		g.traverse_edge("get_flow_dummy", edge_values);
		
		for(Double f:edge_values) {
		
			System.out.println("edge: "+f);
		
		}		
	
	}
	
*/

	
}

